package com.example.demo;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Scheduler;
import io.reactivex.rxjava3.disposables.CompositeDisposable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import org.jetbrains.annotations.NotNull;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;

public class RxDemo {

    OkHttpClient client = new OkHttpClient();

    Observable<String> fetchUrl(String url) {
        return Observable.create(emitter -> {
            Request request = new Request.Builder().url(url).build();
            client.newCall(request).enqueue(new Callback() {
                @Override
                public void onFailure(Call call, IOException e) {
                    emitter.onError(e);
                }

                @Override
                public void onResponse(Call call, Response response) throws IOException {
                    if (response.isSuccessful() && response.body() != null) {
                        String s = response.body().string();
                        System.out.println(s);
                        emitter.onNext(s);
                        emitter.onComplete();
                    } else {
                        emitter.onError(new IOException("Unexpected code " + response));
                    }
                }
            });
        });
    }

    public void call() {
        fetchUrl(url("1", ""))
                .subscribeOn(Schedulers.io())
                .flatMap((s) -> {return fetchUrl(url("2", s));})
                .flatMap((s) -> {return fetchUrl(url("3", s));})
                .doFinally(()->{
                    client.dispatcher().executorService().shutdown();
                    Schedulers.shutdown();
                })
                .blockingSubscribe(
                        result -> System.out.println(result),
                        error -> {
                            error.printStackTrace();
                            System.err.println(error.getMessage());
                        }
                );
    }

    public void call2() {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            return fetchData(url("1", ""));
        });
        CompletableFuture<String> f2 = f1.thenCompose(result1 -> CompletableFuture.supplyAsync(() -> {
            return fetchData(url("2", result1));
        }));
        CompletableFuture<String> future3 = f2.thenCompose(result2 -> CompletableFuture.supplyAsync(() -> {
            return fetchData(url("3", result2));
        }));

        // 等待所有任务完成，并打印最终结果
        future3.thenAccept(finalResult -> {
            System.out.println("Final Result: " + finalResult);
            client.dispatcher().executorService().shutdown();
        }).join(); // 阻塞主线程直到任务完成

    }

    public static void main(String[] args) {
        new RxDemo().call2();
    }

    private String url(String s, String s2) {
        int len = s2.length() >= 30 ? 30 : s2.length();
        return "http://httpbin.org/get?param1=" + s + "&param2=" + s2.substring(0, len);
    }

    void fetchData(String url, Callback callback) {
        Request request = new Request.Builder().url(url).build();
        client.newCall(request).enqueue(callback);
    }

    String fetchData(String url) {
        final var holder = new String[]{""};
        CountDownLatch latch = new CountDownLatch(1);
        fetchData(url, new Callback() {
            @Override
            public void onFailure(@NotNull Call call, @NotNull IOException e) {
                latch.countDown();
            }

            @Override
            public void onResponse(@NotNull Call call, @NotNull Response response) throws IOException {
                holder[0] = response.body().string();
                latch.countDown();
            }
        });
        try {
            latch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return holder[0];
    }
}